package org.baojie.raft.future;

import org.baojie.raft.delay.DelayKey;
import org.baojie.raft.util.current.RaftFactory;
import org.baojie.raft.util.current.RaftThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

public class FutureExecutor {

    private static final Logger log = LoggerFactory.getLogger(FutureExecutor.class);

    private final DelayQueue<DelayKey<Future<?>>> delays = new DelayQueue<>();

    private final RaftThreadPool executor;

    private volatile boolean stop = false;

    private final Thread monitor = new Thread(new Runnable() {
        @Override
        public void run() {
            for (; ; ) {
                if (stop) {
                    return;
                } else {
                    try {
                        DelayKey<Future<?>> entity = delays.poll(45, TimeUnit.MINUTES);
                        if (null != entity) {
                            Future<?> future = entity.getKey();
                            if (null == future) {
                                future.cancel(true);
                            }
                        } else {
                            log.warn("i am future executor monitor, is working now ...");
                        }
                    } catch (Throwable t) {
                        if (stop) {
                            return;
                        } else {
                            log.error("future executor monitor error = " + t.getMessage(), t);
                        }
                    }
                }
            }
        }
    }, "future executor monitor");

    private final Thread hook = new Thread(new Runnable() {
        @Override
        public void run() {
            stop();
        }
    }, "shut down future pool hook");


    private FutureExecutor() {
        final ThreadFactory factory = RaftFactory.create("do raft future work thread", false);
        this.executor = new RaftThreadPool(4, 512, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), factory);
        Runtime.getRuntime().addShutdownHook(hook);
    }

    private static volatile FutureExecutor instance;

    public static FutureExecutor getInstance() {
        if (null == instance) {
            synchronized (FutureExecutor.class) {
                if (null == instance) {
                    instance = new FutureExecutor();
                }
            }
        }
        return instance;
    }

    public Future<?> submit(Runnable runnable, long timeOut, TimeUnit unit) {
        if (null == runnable || null == unit) {
            throw new NullPointerException();
        } else if (null == unit) {
            throw new NullPointerException();
        } else if (timeOut < 0) {
            throw new IllegalArgumentException();
        } else {
            Future<?> future = executor.submit(runnable);
            DelayKey<Future<?>> delay = new DelayKey<>(future, timeOut, unit);
            delays.offer(delay);
            return future;
        }
    }

    public void stop() {
        if (stop) {
            return;
        } else {
            synchronized (FutureExecutor.class) {
                if (!stop) {
                    stop = true;
                    delays.clear();
                    monitor.interrupt();
                    RaftThreadPool tmp = executor;
                    if (null != tmp) {
                        tmp.shutdown(tmp);
                    }
                }
            }
        }
    }

}
